package com.neusoft.databus.agent.interactive.collect;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Arrays;
import java.util.Map.Entry;
import java.util.Optional;

import com.google.common.collect.Multimap;
import com.neusoft.bizcore.web.utils.SpringUtil;
import com.neusoft.databus.agent.parser.modbus.ModBusParser;
import com.neusoft.databus.agent.pool.serial.SerialConnector;
import com.neusoft.databus.common.bean.CommandBean;
import com.neusoft.databus.common.bean.MetricBean;
import com.neusoft.databus.common.connection.SerialConnection;
import com.neusoft.databus.common.exception.RetryableException;
import com.neusoft.databus.common.parser.Parser;
import com.neusoft.databus.common.utils.DigitalUtils;

import gnu.io.SerialPort;
import gnu.io.UnsupportedCommOperationException;
import lombok.extern.slf4j.Slf4j;

/**
 * modbus collector through sirial
 *
 * @author sunchf
 * @date 2019年1月4日 下午5:31:26
 */
@Slf4j
public class SerialModbusCollector extends BasicCollector<SerialPort> {

    public SerialModbusCollector(final Parser parser) {
        super(parser);
        this.connector = SpringUtil.getBean(SerialConnector.class);
    }

    @Override
    public void sendCommand(final CommandBean command) {

        if (SerialModbusCollector.log.isDebugEnabled()) {
            SerialModbusCollector.log.debug("begin to communicate with serial port，{} metrics will be executed",
                    command.getResourceBean().stream().mapToInt(it -> it.getMetrics().size()).sum());
        }

        final ModBusParser mparser = (ModBusParser) this.parser;
        final Multimap<String, MetricBean> multimap = mparser.parseCollectRequest(command);

        if (SerialModbusCollector.log.isDebugEnabled()) {
            SerialModbusCollector.log.debug("after optimizing, it will communicate with serial port {} times",
                    multimap.keySet().size());
        }

        final int totalTimeout = ((SerialConnection) command.getConnection()).getSotimeout() * 1000;
        final int onceTimeout = (totalTimeout / multimap.keySet().size());
        try {
            // it is not effect in windows platform, linux maybe ok, not tested all.
            this.conn.enableReceiveTimeout(onceTimeout);
        } catch (final UnsupportedCommOperationException e) {
            SerialModbusCollector.log.warn("", e);
        }
        final int retry = ((SerialConnection) command.getConnection()).getRetry();

        if (SerialModbusCollector.log.isDebugEnabled()) {
            SerialModbusCollector.log.debug("total timeout: {}, once timeout: {}, retry: {}", totalTimeout, onceTimeout,
                    retry);
        }

        InputStream is = null;
        OutputStream os = null;
        try {
            is = this.conn.getInputStream();
            os = this.conn.getOutputStream();
        } catch (final IOException e) {
            try {
                this.conn.close();
            } catch (final Exception e1) {
            }
            throw new RetryableException(e.getMessage(), e);
        }

        for (final String req : multimap.keySet()) {
            final long startTime = System.currentTimeMillis();
            if (SerialModbusCollector.log.isDebugEnabled()) {
                SerialModbusCollector.log.debug("send a request: {}", req);
            }

            byte[] collectedData = null;
            try {
                collectedData = this.retrieve(req, is, os, onceTimeout, retry);
            } catch (final IOException e) {
                try {
                    this.conn.close();
                } catch (final Exception e1) {
                }
                throw new RetryableException(e.getMessage(), e);
            }
            if (null == collectedData) {
                SerialModbusCollector.log.warn("do not recevice any excepted response.");
                // 一个指令没成功，继续其它的指令
                continue;
            }

            if (SerialModbusCollector.log.isDebugEnabled()) {
                SerialModbusCollector.log.debug("receive a response {}，elapsed {} ms.",
                        DigitalUtils.bytesToHexString(collectedData), System.currentTimeMillis() - startTime);
            } else if (SerialModbusCollector.log.isInfoEnabled()) {
                SerialModbusCollector.log.info("receive a response，elapsed {} ms.",
                        System.currentTimeMillis() - startTime);
            }

            for (final MetricBean metricBean : multimap.get(req)) {
                metricBean.setOriginalValue(collectedData);
            }
        }

        if (SerialModbusCollector.log.isDebugEnabled()) {
            SerialModbusCollector.log.debug("end of communication with serial port，{} metric data received.",
                    command.getResourceBean().stream()
                            .mapToLong(it -> it.getMetrics().stream().filter(m -> m.isExecutedSuccess()).count())
                            .sum());
        }
    }

    /**
     * send and receive data
     *
     * @param req
     * @param is
     * @param os
     * @param retry
     * @return byte[]
     * @throws IOException
     * @throws InterruptedException
     */
    public byte[] retrieve(final String req, final InputStream is, final OutputStream os, final int timeout, int retry)
            throws IOException {

        if (retry < 0) {
            return null;
        }
        final byte[] adu = DigitalUtils.hexStringToBytes(req);

        if (is.available() > 0) {
            final int length = is.available();
            final long n = this.skip(is, length);
            if (SerialModbusCollector.log.isWarnEnabled()) {
                SerialModbusCollector.log.warn("{} bytes are availabe in input stream, {} bytes were skipped.", length,
                        n);
            }
        }

        os.write(adu);
        os.flush();
        final byte[] buf = new byte[2048];
        final int expectedResDataLength = ((ModBusParser) this.parser).expectedResDataLength(adu);
        final long readStartAt = System.currentTimeMillis();
        final long waitTimeEnd = readStartAt + timeout;

        if (SerialModbusCollector.log.isDebugEnabled()) {
            SerialModbusCollector.log.debug("max timeout is {}ms, and expected response byte length is {}.", timeout,
                    expectedResDataLength);
        }

        while ((is.available() < expectedResDataLength) && (System.currentTimeMillis() < waitTimeEnd)) {
            try {
                Thread.sleep(timeout / 10);
            } catch (final InterruptedException e) {
                SerialModbusCollector.log.error("", e);
            }
        }

        int n = -1;
        if (is.available() > 0) {
            n = is.read(buf, 0, buf.length);
        }

        if (SerialModbusCollector.log.isDebugEnabled()) {
            SerialModbusCollector.log.debug("have read byte length is {}.", n);
        }
        if (n != expectedResDataLength) {
            return this.retrieve(req, is, os, timeout, --retry);
        }

        final byte[] data = Arrays.copyOf(buf, n);
        return data;
    }

    @Override
    public void doControl(final CommandBean command) throws Exception {
        if (SerialModbusCollector.log.isDebugEnabled()) {
            SerialModbusCollector.log.debug("start to control task for {}.",
                    command.getResourceBean().get(0).getName());
            SerialModbusCollector.log.debug("borrow connection object from pool {} by {}", command.getConnType(),
                    command.getConnectionStr());
        }

        try {
            this.conn = this.connector.getConn(command.getConnection().asUsedConnection());
            this.tryControl(command);
        } catch (final RetryableException e) {
            throw e;
        } catch (final Exception e) {
            throw e;
        } finally {
            try {
                if (SerialModbusCollector.log.isDebugEnabled()) {
                    SerialModbusCollector.log.debug("return connection object to poll, {}", command.getConnection());
                }
                this.connector.returnConn(command.getConnection().asUsedConnection(), this.conn);
            } catch (final Exception e) {

            }
        }
    }

    private void tryControl(final CommandBean command) throws IOException {
        final int totalTimeout = ((SerialConnection) command.getConnection()).getSotimeout() * 1000;
        final ModBusParser mparser = (ModBusParser) this.parser;
        final Multimap<String, MetricBean> multimap = mparser.parseCollectRequest(command);

        final Optional<Entry<String, MetricBean>> entry = multimap.entries().stream().findFirst();

        final String req = entry.get().getKey();
        if (SerialModbusCollector.log.isDebugEnabled()) {
            SerialModbusCollector.log.debug("send a control request: {}", req);
        }
        final byte[] response =
                this.control(req, this.conn.getInputStream(), this.conn.getOutputStream(), totalTimeout);
        final String resHex = DigitalUtils.bytesToHexString(response);
        if (SerialModbusCollector.log.isDebugEnabled()) {
            SerialModbusCollector.log.debug("receive a control response: {}", resHex);
        }
        if (req.equalsIgnoreCase(resHex)) {
            entry.get().getValue().setOriginalValue(response);
        }

    }

    /**
     * the logic of control is different from collect.
     *
     * @param req
     * @param is
     * @param os
     * @param timeout
     * @return
     * @throws IOException
     */
    private byte[] control(final String req, final InputStream is, final OutputStream os, final int timeout)
            throws IOException {
        final byte[] adu = DigitalUtils.hexStringToBytes(req);

        if (is.available() > 0) {
            final int length = is.available();
            is.skip(length);
            if (SerialModbusCollector.log.isWarnEnabled()) {
                SerialModbusCollector.log.warn("{} bytes are availabe in inputstream, skip them.", length);
            }
        }

        os.write(adu);
        os.flush();
        final byte[] buf = new byte[1024];
        final long readStartAt = System.currentTimeMillis();
        final long waitTimeEnd = readStartAt + timeout;

        while ((is.available() < adu.length) && (System.currentTimeMillis() < waitTimeEnd)) {
            try {
                Thread.sleep(timeout / 10);
            } catch (final InterruptedException e) {
                SerialModbusCollector.log.error("", e);
            }
        }

        final int n = is.read(buf, 0, buf.length);
        final byte[] data = Arrays.copyOf(buf, n);
        return data;
    }

    /**
     * reference InputStream.skip method, but add log
     *
     * @param is
     * @param n
     * @return
     * @throws IOException
     */
    protected long skip(final InputStream is, final long n) throws IOException {

        final int MAX_SKIP_BUFFER_SIZE = 2048;
        long remaining = n;
        int nr;

        if (n <= 0) {
            return 0;
        }

        final int size = (int) Math.min(MAX_SKIP_BUFFER_SIZE, remaining);
        final byte[] skipBuffer = new byte[size];
        while (remaining > 0) {
            nr = is.read(skipBuffer, 0, (int) Math.min(size, remaining));
            if (nr < 0) {
                break;
            }
            remaining -= nr;
        }

        if (SerialModbusCollector.log.isDebugEnabled()) {
            SerialModbusCollector.log.debug("skip data is {}.", DigitalUtils.bytesToHexString(skipBuffer));
        }

        return n - remaining;
    }

}
